package com.hdu.joinInMemory.factory;


import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.hdu.joinInMemory.annotation.JoinInMemoryConfig;
import com.hdu.joinInMemory.enums.JoinInMemeoryExecutorType;
import com.hdu.joinInMemory.executor.JoinItemExecutor;
import com.hdu.joinInMemory.executor.JoinItemsExecutor;
import com.hdu.joinInMemory.executor.ParallelJoinItemsExecutor;
import com.hdu.joinInMemory.executor.SerialJoinItemsExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

import static java.util.Optional.ofNullable;


@Slf4j
public class DefaultJoinItemsExecutorFactory implements JoinItemsExecutorFactory {
    private final List<JoinItemExecutorFactory> joinItemExecutorFactories;
    private final Map<String, ExecutorService> executorServiceMap;

    public DefaultJoinItemsExecutorFactory(Collection<? extends JoinItemExecutorFactory> joinItemExecutorFactories,
                                           Map<String, ExecutorService> executorServiceMap) {
        this.joinItemExecutorFactories = Lists.newArrayList(joinItemExecutorFactories);

        // 按执行顺序进行排序
        AnnotationAwareOrderComparator.sort(this.joinItemExecutorFactories);
        this.executorServiceMap = executorServiceMap;
    }


    @Override
    public <D> JoinItemsExecutor<D> createFor(Class<D> cls) {
        List<JoinItemExecutor<D>> joinItemExecutors = this.joinItemExecutorFactories.stream()
                .flatMap(factory -> factory.createForType(cls).stream())
                .collect(Collectors.toList());
        return buildJoinItemsExecutor(cls, cls.getAnnotation(JoinInMemoryConfig.class), joinItemExecutors);
    }

    private <D> JoinItemsExecutor<D> buildJoinItemsExecutor(Class<D> cls,
                                                            JoinInMemoryConfig joinInMemoryConfig,
                                                            List<JoinItemExecutor<D>> joinItemExecutors) {
        return ofNullable(joinInMemoryConfig)
                .map(config -> {
                    switch (config.executorType()) {
                        case SERIAL:
                            log.info("JoinInMemory for {} use serial executor", cls);
                            return new SerialJoinItemsExecutor<>(cls, joinItemExecutors);
                        case PARALLEL:
                            log.info("JoinInMemory for {} use parallel executor, pool is {}", cls, config.executorName());
                            ExecutorService executor = executorServiceMap.get(config.executorName());
                            Preconditions.checkArgument(executor != null);
                            return new ParallelJoinItemsExecutor<>(cls, joinItemExecutors, executor);
                        default:
                            throw new IllegalArgumentException();
                    }
                })
                .orElse(new SerialJoinItemsExecutor<>(cls, joinItemExecutors));
    }
}
